package zisu.netty.practice.RPC.client;

import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import zisu.netty.practice.RPC.client.codec.RpcRequest;
import zisu.netty.practice.RPC.client.codec.RpcResponse;
import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Desc: "业务处理器"
 * @Author: caixiang
 * @DATE: 2021/1/12 13:59
 */

//SimpleChannelInboundHandler 会帮你自动释放资源
public class RpcClientHandler extends SimpleChannelInboundHandler<RpcResponse> {
    //实际的业务处理类 也缓存 了 这个连接的channel 和remotePeer，供外部使用。
    private Channel channel;
    private SocketAddress remotePeer;

    //pendingRpcTable 是作为一个缓冲的，请求过来以后就放到这个Map里
    private Map<String /* requestId */,RpcFuture> pendingRpcTable = new ConcurrentHashMap<>();

    public Channel getChannel() {
        return channel;
    }

    public SocketAddress getRemotePeer() {
        return remotePeer;
    }

    //通道 注册的时候 调用这个方法（此方法发生在channelActive 之前）
    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
        //在通道注册好的时候 ，拿到 channel。
        this.channel = ctx.channel();
    }

    //通道激活时(激活的意思是真正有连接了) 触发此方法
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        //在激活的时候 就知道 我和哪个远程服务建立连接了。
        this.remotePeer = this.channel.remoteAddress();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcResponse response) throws Exception {
        String requestId = response.getRequestId();
        RpcFuture rpcFuture = pendingRpcTable.get(requestId);
        if(rpcFuture!=null){
            pendingRpcTable.remove(requestId);
            rpcFuture.done(response);
        }
    }

    /**
     * Netty提供了一种主动关闭连接的方式,发送一个Unpooled.EMPTY_BUFFER,这样ChannelFutureListener的close事件就会监听到并关闭通道
     * */
    public void close() {
        //writeAndFlush(Unpooled.EMPTY_BUFFER) channel 主动的发送一个空的buffer；；
        // 空的buffer发出去以后就能被 ChannelFutureListener.CLOSE 监听到
        channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
    }

    /**
     * 异步发送请求方法
     * */
    public RpcFuture sendRequest(RpcRequest rpcRequest){
        //把请求封装成 一个future模型，这样外部就可以addListener了。
        RpcFuture rpcFuture = new RpcFuture(rpcRequest);
        pendingRpcTable.put(rpcRequest.getRequestId(), rpcFuture);
        channel.writeAndFlush(rpcRequest).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                System.out.println("client端，发送请求成功, requestId是："+rpcRequest.getRequestId());
            }
        });
        return rpcFuture;
    }

}